<?php

namespace App\Common\Queue;

use App\Extend\SysExtend\Helper;
use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Queue\Job;

/**
 * 消费者示例
 * Class QueueConsumerEq
 * @package App\Common\Queue
 */
class QueueConsumerEq extends AbstractProcess
{

    /**
     * 执行逻辑
     * @param $arg
     */
    protected function run($arg)
    {
        go(function () {
            if(!Helper::Queue()->getPid('queue_consumer_eq')) {
                // 将进程ID存入缓存
                Helper::Queue()->setPid('queue_consumer_eq', $this->getPid());
            }
            Helper::Queue()->getQueue('queue_consumer_eq')->consumer()->listen(function (Job $job){
                echo 'queue_consumer_eq:'.json_encode($job->toArray()).PHP_EOL;
            });
        });
    }

    /**
     * 异常回调
     * @param \Throwable $throwable
     * @param mixed ...$args
     * @throws \Throwable
     */
    protected function onException(\Throwable $throwable, ...$args)
    {
        parent::onException($throwable,...$args);
    }
}